package com.stream.consumer.listener;

import com.stream.consumer.MySink;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class ConsumerListener {

    // 只接收头信息 test-header = my test 的消息
    @StreamListener(value = Sink.INPUT,condition = "headers['test-header']=='my test'")
    public void receive(String msg){
        System.out.println("receive msg: " + msg);
    }

    @StreamListener(value = MySink.INPUT)
    public void receiveCustorm(String msg){
        System.out.println("MySink receive msg: " + msg);

        throw new IllegalArgumentException("异常测试");
    }

    @StreamListener("errorChannel") // errorChannels 固定不能变
    public void handleError(ErrorMessage errorMessage){
        log.error("全局异常: {}",errorMessage);
    }

    // 局部异常   // inputChannel = 通道 + 组名 + errors(固定写法)
    /*@ServiceActivator(
        inputChannel = "topic-test-stream-myouput.my-stream-consumer-group.errors"
    )
    public void handleError(ErrorMessage errorMessage){
        log.error("局部异常: {}",errorMessage);
    }*/
}
